package com.atguigu.flink.chapter11;

import com.atguigu.flink.chapter05.Source.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.rowInterval;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/1/27 14:19
 */
public class Flink13_SQL_GroupWindow {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);


        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("create table sensor(" +
                "id string," +
                "ts bigint," +
                "vc int, " +
                "t as to_timestamp(from_unixtime(ts/1000,'yyyy-MM-dd HH:mm:ss'))," +
                "watermark for t as t - interval '5' second)" +
                "with("
                + "'connector' = 'filesystem',"
                + "'path' = 'input/sensor-sql.txt',"
                + "'format' = 'csv'"
                + ")"
        );

        tableEnv
                .executeSql("SELECT id, " +
                        "  TUMBLE_START(t, INTERVAL '1' minute) as wStart,  " +
                        "  TUMBLE_END(t, INTERVAL '1' minute) as wEnd,  " +
                        "  SUM(vc) sum_vc " +
                        "FROM sensor " +
                        "GROUP BY TUMBLE(t, INTERVAL '1' minute), id"
                )
                .print();

    }
}
/*
    SQL里使用GroupWindow：
        1、调用函数开窗
            TUMBLE（时间字段，窗口长度）    窗口长度格式： interval 'n' 时间单位
            HOP（时间字段，滑动步长，窗口长度）     窗口长度、步长格式： interval 'n' 时间单位
        2、开窗语句要放在groupby 中

        3、通过其他函数获取窗口信息：
            TUMBLE_START(跟开窗一样)     窗口开始时间
            TUMBLE_END(跟开窗一样)       窗口结束时间
 */